from confluent_kafka import Producer, Consumer
from confluent_kafka.admin import AdminClient, NewTopic, OffsetSpec
from confluent_kafka import KafkaException

# 生产者示例
from confluent_kafka.cimpl import TopicPartition


def produce_message(topic, message):
    # 配置Kafka生产者
    producer_conf = {
        'bootstrap.servers': 'kafka1:9092,kafka2:9092',  # Kafka集群的地址
        # 可选配置
        # 'security.protocol': 'SASL_SSL',
        # 'sasl.mechanisms': 'PLAIN',
        # 'sasl.username': 'your_username',
        # 'sasl.password': 'your_password',
    }

    producer = Producer(producer_conf)

    # 发送消息到指定主题
    try:
        producer.produce(topic, message.encode('utf-8'))
        producer.flush()  # 等待所有消息发送完成
        print(f"Produced message: {message} to topic {topic}")
    except Exception as e:
        print(f"Failed to produce message: {e}")
    finally:
        producer.close()  # 关闭生产者

# 消费者示例
def consume_messages(topic):
    # 配置Kafka消费者
    consumer_conf = {
        'bootstrap.servers': 'kafka1:9092,kafka2:9092',  # Kafka集群的地址
        'group.id': 'my_consumer_group',  # 消费者组ID
        'auto.offset.reset': 'earliest'  # 从最早的消息开始消费
        # 可选配置
        # 'security.protocol': 'SASL_SSL',
        # 'sasl.mechanisms': 'PLAIN',
        # 'sasl.username': 'your_username',
        # 'sasl.password': 'your_password',
    }

    consumer = Consumer(consumer_conf)

    # 订阅主题并开始消费消息
    consumer.subscribe([topic])

    try:
        while True:
            msg = consumer.poll(timeout=1.0)  # 消费消息
            if msg is None:
                continue
            if msg.error():
                print(f"Consumer error: {msg.error()}")
                continue

            # 处理收到的消息
            print(f"Consumed message: {msg.value().decode('utf-8')} from topic {msg.topic()}")

    except KeyboardInterrupt:
        pass
    finally:
        consumer.close()  # 关闭消费者

# 查询指定topic的数据
def query_kafka_topic_metadata(bootstrap_servers, topic_name):
    admin_conf = {'bootstrap.servers': bootstrap_servers}
    admin_client = AdminClient(admin_conf)

    try:
        # 查询主题列表
        topic_metadata = admin_client.list_topics(timeout=10)
        if topic_name in topic_metadata.topics:
            topic_info = topic_metadata.topics[topic_name]
            print(f"Topic '{topic_name}' found with partitions: {topic_info.partitions}")
            for partition_id, partition in topic_info.partitions.items():
                print(f"  Partition {partition_id}: Leader={partition.leader}, Replicas={partition.replicas}")

                # 查询每个分区的最新偏移量
                fetch_partition_offsets(admin_conf, topic_name, partition.id)

            # 获取主题的配置信息
            config_entries = admin_client.describe_configs([topic_name], timeout=10)
            for resource, config in config_entries.items():
                print(f"Topic '{topic_name}' config: {config}")

        else:
            print(f"Topic '{topic_name}' not found.")

    except KafkaException as e:
        print(f"Kafka error: {e}")
    # finally:
    #     admin_client.close()

# 查询每个分区的最新偏移量
def fetch_partition_offsets(admin_conf, topic_name, partition_id):
    admin_client = AdminClient(admin_conf)

    try:
        # 获取主题的分区信息
        partitions = {TopicPartition(topic_name, partition_id): OffsetSpec.latest()}
        offsets = admin_client.list_offsets(partitions)

        for partition, offset in offsets.items():
            if offset.offset is None:
                print(f"  Partition {partition.partition}: Offset not available")
            else:
                print(f"  Partition {partition.partition}: Latest offset={offset.offset}")

    except KafkaException as e:
        print(f"Kafka error: {e}")


# 示例用法    ----待测试
bootstrap_servers = '192.168.101.200:9092'  # Kafka 集群的地址
topic_name = 'objects_list'  # 要查询的主题名称
query_kafka_topic_metadata(bootstrap_servers, topic_name)



# 使用示例
# if __name__ == "__main__":
#     # 替换为你的Kafka主题名称
#     topic_name = "my_topic"
#
#     # 生产者发送消息到主题
#     produce_message(topic_name, "Hello Kafka!")
#
#     # 消费者消费消息
#     consume_messages(topic_name)